package timeandwindow;

/*
*熟练使用窗口的前置结论：

*   1.数据落哪个窗口，看的是数据本身的时间
*
*   2.时间窗口的计算，看的是事件，如果是处理时间语义，直接看系统时间，如果是事件时间语义，看水位线
*
*   3.窗口是动态创建的，当有数据需要落到该窗口的时候，会创建该窗口（先创建窗口，后触发计算）
*
*   4.窗口计算输出结果 和 窗口关闭是两个行为，默认窗口计算输出结果，直接关闭窗口
*
*   5.使用窗口的时候，重点关注两件事：
*      1）窗口分配器，决定了使用哪种类型的窗口
*      2）窗口函数：决定了窗口中的数据如何进行计算
*
*   窗口分配：
*      如果是处理时间语义的窗口，按照数据到达的时候，获取一次当前的处理时间（系统时间），通过处理时间计算窗口
*      如果是事件时间语义窗口，按照数据中的时间，通过事件时间计算窗口
*
*   按键分区窗口：
*      在开窗口之前，会进行keyBy操作，每个key的数据都会对应自己独有的窗口，窗口也是按照key隔离的
*
*
*
* */

import com.atguigu.pojo.Event;
import com.atguigu.pojo.WordCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Flink05_keyedWindows {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);

         DataStreamSource<String> ds =
                 env.socketTextStream("hadoop102", 9999);

         //对接数据源
         SingleOutputStreamOperator<Event> ds1 = ds.map(
                 line -> {
                     String[] fileds = line.split(",");
                     return new Event(fileds[0].trim(), fileds[1].trim(), Long.valueOf(fileds[2].trim()));
                 }
         ).assignTimestampsAndWatermarks(
                 WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)//延迟时间为2s
                         .withTimestampAssigner(
                                 (element, ts) -> element.getTs() //返回事件时间戳
                         )
         );

         ds1.print("INPUT");

        //按键分区窗口
            //1.计数窗口
            //1.1计数滚动窗口
         //ds1.map(event -> new WordCount(event.getUser(),1L))
         //        .keyBy(
         //                WordCount::getWord
         //        )
         //        .countWindow(3L)
         //        .sum("count")
         //        .print();
         //  //1.2计数滑动窗口
         //ds1.map(event -> new WordCount(event.getUser(),1L))
         //        .keyBy(WordCount::getWord)
         //        .countWindow(3L,2L)
         //        .sum("count")
         //        .print();

         //时间窗口
         ds1.map( event -> new WordCount(event.getUser(),1L))
                 .keyBy(
                         WordCount::getWord
                 ).window(
                  //时间滚动窗口
                      //处理时间
                         //TumblingProcessingTimeWindows.of(Time.seconds(5))
                      //事件时间
                        // TumblingEventTimeWindows.of(Time.seconds(3))
                 //时间滑动窗口
                      //处理时间
                        // SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5))
                      //事件时间
                        // SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5))
                 //时间会话窗口
                      //处理时间
                       //  ProcessingTimeSessionWindows.withGap(Time.seconds(5))
                      //事件时间 todo 注意必须是相邻会话间隔超过5秒或者即将超过5秒
                         EventTimeSessionWindows.withGap(Time.seconds(5))
                 )
                 .sum("count")
                 .print();
            //
     //按照key数数即可，条数够了触发计算
         //按照key隔离开看效果
         //时间窗口和计数窗口不同，时间到了触发窗口计算，不同key都要计算

             try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}